package com.xy.iot.netty;

import com.xy.iot.common.message.BaseMessage;
import com.xy.iot.netty.channel.BaseChannel;
import com.xy.iot.netty.channel.BaseChannelGroup;
import com.xy.iot.netty.connect.BaseServerConnect;
import com.xy.iot.netty.device.BaseControlCentre;
import com.xy.iot.netty.device.DeviceServerChannel;
import com.xy.iot.netty.handle.ByteServerMessageDecoder;
import com.xy.iot.netty.handle.ByteServerMessageEncoder;
import com.xy.iot.netty.handle.DelimiterMessageDecoder;
import com.xy.iot.netty.handle.ExceptionServerHandler;
import com.xy.iot.netty.listener.Listener;
import com.xy.iot.netty.listener.ListenerServerHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 服务启动
 * 案例：
 *      <a href="https://blog.csdn.net/qq_28369007/article/details/136307226">...</a>
 */
@Slf4j
public class StartServer extends ChannelInitializer<SocketChannel> implements BaseControlCentre {
    /**
     * bossGroup 线程组 用于接受进来的连接
     */
    private EventLoopGroup bossGroup;
    /**
     * workerGroup 工作线程组，用于进行SocketChannel的网络读写操作
     */
    private EventLoopGroup workerGroup;
    /**
     * 启动服务
     */
    private ChannelFuture future;
    /**
     * 服务端口
     */
    private final int port;
    /**
     * 线程数,当为0时，默认为CPU核数的2倍
     */
    private final int threadNum;
    /**
     * 心跳检测超时时间(单位：毫秒)
     */
    @Getter
    private final Integer idleTimeout;
    private final Class<? extends BaseMessage> baseMessageData;
    /**
     * 分组通讯通道
     */
    @Getter
    private final BaseChannelGroup groups;
    /**
     * 处理连接成功、断开连接
     */
    private final BaseServerConnect connect;
    private final ListenerServerHandler handler;

    /**
     *  创建指定服务端口，默认线程数，当为0时，默认为CPU核数的2倍
     * @param port 服务端口
     * @param baseMessageData 解析数据源
     */
    public StartServer(int port, Class<? extends BaseMessage> baseMessageData){
        this(port, 0, 60000, baseMessageData);
    }
    /**
     *
     * 创建指定服务端口，指定线程数的服务端
     * @param port 服务端口
     * @param nThread 执行线程池线程数
     * @param heartBeatTime 心跳检测超时时间(单位：毫秒)
     * @param baseMessageData 解析数据源
     */
    public StartServer(int port, int nThread, int heartBeatTime, Class<? extends BaseMessage> baseMessageData){
        this.port = port;
        this.threadNum = nThread;
        this.idleTimeout = heartBeatTime;
        this.baseMessageData = baseMessageData;
        this.groups = new BaseChannelGroup();
        this.connect = new BaseServerConnect();
        this.handler = new ListenerServerHandler(this.groups);
    }

    /**
     * 启动服务
     *         Reactor线程模型
     *              单线程模型：
     *                          EventLoopGroup bossGroup = new NioEventLoopGroup(1);
     *                          ServerBootstrap b = new ServerBootstrap();
     *                          b.group(bossGroup).channel(NioServerSocketChannel.class);
     *              多线程模型：
     *                          EventLoopGroup bossGroup = new NioEventLoopGroup(1);
     *                          EventLoopGroup workerGroup = new NioEventLoopGroup();
     *                          ServerBootstrap b = new ServerBootstrap();
     *                          b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);
     *              主从Reactor线程模型
     *                          EventLoopGroup bossGroup = new NioEventLoopGroup(4);
     *                          EventLoopGroup workerGroup = new NioEventLoopGroup();
     *                          ServerBootstrap b = new ServerBootstrap();
     *                          b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);
     *         channel通道类型
     *                          NioServerSocketChannel: NIO模型
     *                          OioServerSocketChannel：BIO模型
     *                          EpollServerSocketChannel：Linux下使用Epoll模型
     *                          KQueueServerSocketChannel：Mac下使用KQueue模型
     *         option：作用于每个新建立的channel，设置TCP连接中的一些参数，option主要是针对boss线程组
     *
     *                          ChannelOption.SO_BACKLOG, 1024
     *                                                  BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，
     *                                                  用于临时存放已完成三次握手的请求的队列的最大长度。
     *                                                  如果未设置或所设置的值小于1，Java将使用默认值50。
     *
     *                         ChannelOption.SO_KEEPALIVE, true
     *                                                  是否启用心跳保活机制。
     *                                                  在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）
     *                                                  并且在两个小时左右上层没有任何数据传输的情况下，
     *                                                  这套机制才会被激活。
     *
     *                         ChannelOption.TCP_NODELAY, true
     *                                                  是否启用Nagle算法，true表示关闭，false表示启用
     *                                                   在TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，
     *                                                   同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，
     *                                                   TCP总是希望尽可能的发送足够大的数据。这里就涉及到一个名为Nagle的算法，
     *                                                   该算法的目的就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。
     *                       ChannelOption.SO_REUSEADDR, true
     *                                      SO_REUSEADDR允许启动一个监听服务器并捆绑其众所周知端口，即使以前建立的将此端口用做他们的本地端口的连接仍存在。这通常是重启监听服务器时出现，若不设置此选项，则bind时将出错。
     *                                      SO_REUSEADDR允许在同一端口上启动同一服务器的多个实例，只要每个实例捆绑一个不同的本地IP地址即可。对于TCP，我们根本不可能启动捆绑相同IP地址和相同端口号的多个服务器。
     *                                      SO_REUSEADDR允许单个进程捆绑同一端口到多个套接口上，只要每个捆绑指定不同的本地IP地址即可。这一般不用于TCP服务器。
     *                                      SO_REUSEADDR允许完全重复的捆绑：当一个IP地址和端口绑定到某个套接口上时，还允许此IP地址和端口捆绑到另一个套接口上。一般来说，这个特性仅在支持多播的系统上才有，而且只对UDP套接口而言（TCP不支持多播）
     *                       ChannelOption.SO_RCVBUF  AND  ChannelOption.SO_SNDBUF
     *                                          定义接收或者传输的系统缓冲区buf的大小，
     *         childOption：作用于每个新建立的channel，设置TCP连接中的一些参数，主要是针对worker线程组
     *                         选择解释同《option》
     *         handler：用于对服务端通道进行初始化(在初始化时就会执行)，作用于被accept之后的连接
     *         childHandler：用于对每个通道里面的数据处理(会在客户端成功connect后才执行)
     *         bind：绑定端口
     *
     *
     */
    public void start(){
        //设置 Reactor线程模型，为多线程模型
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup(threadNum);
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup)//设置分组
                .channel(NioServerSocketChannel.class)//设置channel通道类型
                .option(ChannelOption.SO_BACKLOG, 5)//backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
                // .option(ChannelOption.SO_KEEPALIVE,true)
                .handler(new LoggingHandler(LogLevel.DEBUG))//设置日志
                .childHandler(this)//设置客户端连接后的处理
        ;
        future = serverBootstrap.bind(port);//绑定端口
        future.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("Netty服务启动完成，工作线程池：{}  已经在端口：{} 进行阻塞等待", threadNum, port);
            } else {
                log.error("Netty启动异常：{}", future.cause().getMessage());
            }
        });
    }

    /**
     * 停止服务
     */
    public void stop(){
        future.channel().closeFuture();
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
        log.info("Netty关闭成功，工作线程池：{}  已经在端口：{}", threadNum, port);
    }

    /**
     * 根据名称/地址找已连接设备组
     * 名称/地址不存在或者未连接时返回null值
     * @param name 名称/地址
     * @return 设备组
     */
    public BaseChannel group(String name) {
        return get(name);
    }

    /**
     * 列出所有已连接设备组清单
     * @return 所有已连接身边组清单
     */
    @Override
    public List<BaseChannel> groups() {
        return groups.list();
    }

    /**
     * 添加接收指定指令的消息监听器
     * @param command 指令类型
     * @param listener 消息监听器
     * @return 上一个消息监听器，如果没有返回null
     */
    public Listener addListener(int command, Listener listener){
        return handler.addListener(command, listener);
    }

    /**
     * 移除接收指定指令的消息监听器
     * @param command 指令类型
     * @return 移除消息监听器，如果没有返回null
     */
    public Listener removeListener(int command){
        return handler.removeListener(command);
    }

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline p = ch.pipeline();
        BaseChannel group = new DeviceServerChannel(ch);
        add(group);
        //服务端心跳检测超时时间，超时则主动断开链接
        p.addLast(new IdleStateHandler(0, 0, idleTimeout, TimeUnit.MILLISECONDS));
        p.addLast(new ChannelInboundHandlerAdapter(){

            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                if(evt instanceof IdleStateEvent){
                    ctx.disconnect();//如果事件确实是IdleStateEvent，则调用ChannelHandlerContext的disconnect()方法断开当前的连接
                } else {
                    super.userEventTriggered(ctx, evt);//如果事件不是IdleStateEvent，则调用父类的userEventTriggered方法来继续传递这个事件给管道中的下一个处理器。这样，非空闲相关的用户事件能够被其他合适的处理器正确处理。
                }
            }
        });
      p.addLast(new ByteServerMessageEncoder());
      p.addLast(new DelimiterMessageDecoder());
      p.addLast(new ByteServerMessageDecoder(baseMessageData));
      p.addLast(connect);
      p.addLast(handler);
      p.addLast(new ExceptionServerHandler());
    }

    @Override
    public boolean add(BaseChannel baseChannel) {
        return groups.add(baseChannel);
    }

    @Override
    public BaseChannel remove(String name) {
        return groups.remove(name);
    }

    @Override
    public BaseChannel get(String name) {
        return groups.get(name);
    }

}

